package org.xurui.sxl.protocol.interceptor;


import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.cache.Caches;
import org.jetlinks.core.device.DeviceOperator;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.message.function.FunctionInvokeMessage;
import org.jetlinks.core.message.function.FunctionParameter;
import org.jetlinks.core.message.interceptor.DeviceMessageSenderInterceptor;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import org.xurui.sxl.protocol.enums.CommandEnum;
import reactor.core.publisher.Flux;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
public class MyDeviceMessageSenderInterceptor implements DeviceMessageSenderInterceptor {
    ConcurrentMap<String, Set<DeviceMessage>> willMsgCache = Caches.newCache();
    DelayQueue<WillSenderDelay> delayQueue = new DelayQueue<WillSenderDelay>();

    public MyDeviceMessageSenderInterceptor(){
        //启动发送离线消息任务
        doSendWillDeviceMsg();
    }

    /**
     * 在消息发送后触发.这里发送后并不是真正的发送，其实只是构造了整个发送的逻辑流{@link Flux}(参数 reply),
     *
     * @param device  设备操作接口
     * @param message 源消息
     * @param reply   回复的消息
     * @param <R>     回复的消息类型
     * @return 新的回复结果
     */
    @Override
     public  <R extends DeviceMessage> Flux<R> afterSent(DeviceOperator device, DeviceMessage message, Flux<R> reply) {
        return reply.onErrorResume(err->{
            addWillDeviceMsg(device,message);
            return Flux.error(err);
        });
    }

    private void addWillDeviceMsg( DeviceOperator device,DeviceMessage message){
        if(eliminateDeviceMsg(message)){
            String key = device.getDeviceId()+":will";
            Set<DeviceMessage> set = willMsgCache.computeIfAbsent(key,(k)->{
                return new HashSet<DeviceMessage>();
            });
            set.add(message);
            log.debug("添加设备补发消息 设备【{}】消息【{}】",device,message);
        }

    }

    private boolean eliminateDeviceMsg(DeviceMessage message){
        if(message instanceof FunctionInvokeMessage){
            FunctionInvokeMessage obj = (FunctionInvokeMessage)message;
            String enumType=obj.getFunctionId();
            if(CommandEnum.R1.name().equals(enumType)){
                return false;
            }else  if(CommandEnum.GJ.name().equals(enumType)){
                List<FunctionParameter> params = obj.getInputs();
                Map<String, Object> paramsMap = params.stream()
                    .collect(Collectors.toMap(FunctionParameter::getName, FunctionParameter::getValue));
                String repeatFlag="";
                if(paramsMap.get("repeatFlag")!=null && !paramsMap.get("repeatFlag").equals("")) {
                    repeatFlag=paramsMap.get("repeatFlag").toString();
                }
                String offTime="";
                if(paramsMap.get("offTime")!=null && !paramsMap.get("offTime").equals("")) {
                    offTime=paramsMap.get("offTime").toString();
                }
                String onTime="";
                if(paramsMap.get("onTime")!=null && !paramsMap.get("onTime").equals("")) {
                    onTime=paramsMap.get("onTime").toString();
                }

                if(!StringUtils.isEmpty(repeatFlag)&& !StringUtils.isEmpty(offTime) && !StringUtils.isEmpty(onTime)){
                    if(repeatFlag.equals("0000000")&&offTime.equals(onTime)){
                        return false;
                    }else {
                        return true;
                    }
                }else {
                    return true;
                }



            }else {
                return true;
            }
        }
        return true;
    }

    public void sendWillDeviceMsgDelay(DeviceOperator device){
        if(device.getDeviceId()!=null
            &&willMsgCache.containsKey(device.getDeviceId()+":will")){
            //五分钟中执行
            WillSenderDelay willSenderDelay = new WillSenderDelay(device,1000*60*3L);
            delayQueue.add(willSenderDelay);
        }

    }

    private void doSendWillDeviceMsg(){
        Runnable runnable = ()->{
            while(true){
                try {
                    log.debug("执行补发设备消息任务");
                    WillSenderDelay willSenderDelay = delayQueue.take();
                    DeviceOperator device = willSenderDelay.deviceOperator;
                    Set<DeviceMessage> set = willMsgCache.remove(device.getDeviceId()+":will");
                    if(!ObjectUtils.isEmpty(set)){
                        Flux.fromIterable(set)
                            .doOnNext(msg->{
                                log.debug("补发设备消息 设备【{}】消息【{}】",device,msg);
                            }).flatMap(msg->{
                            return device.messageSender().sendAndForget(msg);
                        }).subscribe();
                    }
                } catch (Exception e) {
                    log.error("重发离线消息异常",e);
                }
            }

        };
        new Thread(runnable).start();

    }

    class WillSenderDelay implements Delayed{
        Long delayTime;//延迟时间
        Long expire;//过期时间
        DeviceOperator deviceOperator;
        public WillSenderDelay(DeviceOperator deviceOperator,Long delayTime) {
            this.delayTime = delayTime;
            // 过期时间 = 当前时间 + 延迟时间
            this.expire = System.currentTimeMillis() + delayTime;
            this.deviceOperator = deviceOperator;
        }

        /**
         * 剩余时间 = 到期时间 - 当前时间
         */
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        /**
         * 优先级规则：两个任务比较，时间短的优先执行
         */
        @Override
        public int compareTo(Delayed o) {
            long f = this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
            return (int) f;
        }
    }
}
